package ex;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
import org.eclipse.jetty.util.IO;

import java.io.*;
import java.util.Map;

public class SpoutMock extends BaseRichSpout {
    private SpoutOutputCollector collector;
    private BufferedReader bufferedReader; // 字符流，常用

    @Override
    public void open(Map<String, Object> map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        collector = spoutOutputCollector;
//        InputStream inputStream = ClassLoader.getSystemClassLoader().getResourceAsStream("SpoutMockData.txt");
//        assert inputStream != null;
//        bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
        try{
            bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream("/home/liu/Source/BigDataSource/Project/StormDemo/Chapter02/src/main/resources/SpoutMockData.txt")));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void nextTuple() {
        String sentence;

        try {
            while ((sentence = bufferedReader.readLine()) != null) {
                collector.emit(new Values(sentence));
                Utils.sleep(100);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("info"));
    }
}
